﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Diagnostics.Tracing;
using System.IO;
using System.Runtime.Serialization.Formatters.Binary;
using System.Threading.Actors.Runtime;
using System.Threading.Actors.Utilities;
using System.Threading.Tasks;

namespace System.Threading.Actors
{
    public class ReplicatingActorState : StateProviderBase, IPersistenceHandler
    {

        // Local variables that represent in memory state
        //private readonly object operationLockObject = new object();
        private long sequenceNumber = 0;

        // Common logic for multi-buffer state handling
        private readonly StateBufferingHelper m_stateHelper;

        private readonly BinaryFormatter binaryFormatter = new BinaryFormatter(); // used for serialization

        // We need to be able to grab the state associated with an actor, so we maintain the 
        // actor name (m_serviceName) in the actor state, and also maintain a dictionary mapping
        // actor name to actor state.
        internal string m_serviceName;

        // Cache a pre-completed task for "early exit" situations so that we don't
        // have to create a new Task each time.
        private static Task<bool> s_preCompletedTask = Task.FromResult(true);

        private ActorETWEventListener m_ETWListener;

        IsolatedActorState m_defaultIsolatedState;
        // Make this a ConcurrentDictionary so that we can traverse it w/o
        // being concerned about whether it is being simultaneously updated.
        ConcurrentDictionary<string, IsolatedActorState> m_keySpecificIsolatedStates;

        public ReplicatingActorState(ILogger logger, string serviceName)
            : base(logger)
        {
            m_serviceName = serviceName;

            // Kick off real-time logging of Actor ETW events
            m_ETWListener = new ActorETWEventListener(serviceName);
            // Use IActor.Configure to change this at runtime.
            m_ETWListener.EnableEvents(ActorETWEventSource.Log, EventLevel.Informational);
            
            // TODO: It's dangerous to leak "this" pointer from the constructor
            m_stateHelper = new StateBufferingHelper(this);
            var interactor = new FabricActorInteractionHelper();
            m_defaultIsolatedState = new IsolatedActorState(m_stateHelper, "", m_serviceName, m_ETWListener, interactor);
            m_keySpecificIsolatedStates = new ConcurrentDictionary<string,IsolatedActorState>();

        }

        // Retrieve the IsolatedState for the given partitionName, creating one if necessary.
        internal IsolatedActorState GetIsolatedState(string partitionName)
        {
            if (String.IsNullOrEmpty(partitionName)) return m_defaultIsolatedState;
            else
            {
                IsolatedActorState result;
                // I'm still doing double-checked locking here even though
                // m_keySpecificIsolatedStates is a ConcurrentDictionary.
                // That's because I don't trust ConcurrentDictionary.GetOrAdd
                // to do the right thing w/o explicit locking around it.
                if (!m_keySpecificIsolatedStates.TryGetValue(partitionName, out result))
                {
                    lock (m_keySpecificIsolatedStates)
                    {
                        if (!m_keySpecificIsolatedStates.TryGetValue(partitionName, out result))
                        {
                            // IsolatedActorState constructor could possibly have some awaits in it,
                            // so we need to run it on a separate task to "escape" the lock that
                            // we are under.
                            //result = new IsolatedActorState(m_stateHelper, partitionName, m_serviceName);
                            var interactor = new FabricActorInteractionHelper();
                            result = Task.Factory.StartNew(() => 
                            {
                                return new IsolatedActorState(m_stateHelper, partitionName, m_serviceName, m_ETWListener, interactor); 
                            }, TaskCreationOptions.PreferFairness).Result;
                            m_keySpecificIsolatedStates[partitionName] = result;
                        }
                    }
                }

                return result;
            }
        }

        // May need soon.
        //internal bool IsolatedStateExists(string partitionName)
        //{
        //    if (String.IsNullOrEmpty(partitionKey))
        //    {
        //        return true;
        //    }
        //    else if (m_keySpecificIsolatedStates != null)
        //    {
        //        return m_keySpecificIsolatedStates.ContainsKey(partitionName);
        //    }
        //    else return false;
        //}


        public string GetId()
        {
            return m_serviceName;
        }

        public void Configure(string configurationString)
        {
            ConfigurationHelper.Configure(configurationString, m_ETWListener);
        }

        #region Actor Lifetime Methods

        // The service is shutting down; clean up any remaining resources.
        // Disposing this will dispose our state helper.  The assumption is this only happens during service shutdown,
        // and the code in ReplicatingActorServiceBase's OnClose or OnAbort already unhooked our assembly resolve event.
        public void Dispose()
        {
            Console.WriteLine("Actor ({0}) closing... ReplicatingActorState.Dispose() entered.", m_serviceName);
            m_defaultIsolatedState.Dispose();
            foreach (var kvp in m_keySpecificIsolatedStates) kvp.Value.Dispose();
            if (m_stateHelper != null) m_stateHelper.Dispose();
            if (m_ETWListener != null) m_ETWListener.Dispose();
        }

        private void OnPromotionToPrimary_helper(string prefix, IsolatedActorState isolatedState)
        {
            Contract.Requires(isolatedState != null);

            var handlers = isolatedState.GetAllKeysWithPrefix(prefix);
            foreach (var handler in handlers)
            {
                var parts = handler.Split('$');
                Contract.Assert(parts.Length == 3, "OnPromotionToPrimary expected three values in a serialized event handler.  You may not use '$' in an actor name nor in a name of an event handler.");
                var sourceActor = parts[0];
                var eventType = parts[1];
                var partitionName = parts[2] == String.Empty ? null : parts[2];
                var handlerName = (string)isolatedState.Get(prefix + handler);

                Console.WriteLine("--Subscribing to (\"{0}\"{3}, \"{1}\", \"{2}\")", sourceActor, eventType, handlerName, partitionName != null ? ":" + partitionName : null);
                var proxy = isolatedState.GetActorProxy(sourceActor, partitionName);
                var observer = proxy[eventType];
                observer.Subscribe(handlerName);
                //Subscribe(sourceActor, eventType, handlerName);
            }

        }

        // Signifies that our owning actor has been promoted to the primary actor.
        // This means that we need to "re-hydrate" a bunch of subscription logic.
        internal void OnPromotionToPrimary()
        {
            Console.WriteLine("{0}: OnPromotionToPrimary({1}) called...", DateTime.Now, m_serviceName);
            var prefix = "$EVENTHANDLER$";
            OnPromotionToPrimary_helper(prefix, m_defaultIsolatedState);
            foreach (var kvp in m_keySpecificIsolatedStates)
            {
                var isolatedState = kvp.Value;
                OnPromotionToPrimary_helper(prefix, isolatedState);                
            }
            Console.WriteLine("{0}: OnPromotionToPrimary({1}) completed.", DateTime.Now, m_serviceName);
        }
        #endregion Actor Lifetime Methods

        #region StateProviderBase methods
        //
        // StateProviderBase methods
        //

        protected override object ApplyActionOnPrimary(ActionBase action, long sequenceNumber)
        {
            // apply to memory

            // TODO: This is probably overly conservative and I think that we're leaving some perf
            // on the table.  The reason that this lock would seem necessary is that I don't want
            // the actor delegate reading items as they are being updated by the replication task --
            // that could lead to torn reads and major correctness problems.
            //
            // However... in most cases, a competing read from the actor delegate would read from 
            // pendingTransactionHolders[REPLICATING] rather than the persistent store itself,
            // which is being updated here.  The one exception to this is GetAllKeysWithPrefix,
            // which first checks the persistent store.
            //
            // So... it would probably be sufficient to have a dictionaryLockObject that was held
            // during crucial points in GetAllKeysWithPrefix and here.  That would mean that normal
            // actor delegate reads would not need to be locked out during this operation, and
            // you would thereby get better performance.
            lock (this)
            {

                var batchAction = action as BatchAction;
                if (batchAction != null)
                {
                    int activityCount = 0;
                    // Assuming we only have two operation types - Set and Remove
                    if (batchAction.SetKeys != null)
                    {
                        for (int i = 0; i < batchAction.SetKeys.Length; i++)
                        {
                            var partitionName = batchAction.SetPartitionNames[i];
                            var key = batchAction.SetKeys[i];
                            Object value = batchAction.SetValues[i];
                            var isolatedState = GetIsolatedState(partitionName);
                            isolatedState.SetDurableValue(key, value, "primary", scour: true);
                            //Console.WriteLine("ApplyActionOnPrimary: Adding key {0} for partitionName {1}", key, partitionName);
                            activityCount++;
                        }
                    }
                    //else this.Logger.Write(Guid.Empty, "Primary servicing batch {0}: No sets", sequenceNumber);

                    if (batchAction.RemoveKeys != null)
                    {
                        for (int i = 0; i < batchAction.RemoveKeys.Length; i++)
                        {
                            var partitionName = batchAction.RemovePartitionNames[i];
                            String key = batchAction.RemoveKeys[i];
                            //this.Logger.Write(Guid.Empty, "Primary servicing Remove({0}) in batch {1}", key, sequenceNumber);
                            var isolatedState = GetIsolatedState(partitionName);
                            isolatedState.RemoveDurableValue(key);
                            //Console.WriteLine("ApplyActionOnPrimary: Removing key {0} for partitionName {1}", key, partitionName);
                            activityCount++;
                        }
                    }
                    //else this.Logger.Write(Guid.Empty, "Primary servicing batch {0}: No removes", sequenceNumber);

                    this.sequenceNumber = sequenceNumber;
                    return activityCount;
                }

                // FAIL if we come here
                Utility.CodingError("Unknown action asked to be applied");
                throw new InvalidOperationException();
            }
        }

        protected override void ApplyActionOnSecondary(ActionBase action, long sequenceNumber)
        {
            // On a secondary we don't need to take locks to Apply
            // The base class guarantees that operations would be called in sequence

            var batchAction = action as BatchAction;
            if (batchAction != null)
            {
                // Assuming we only have two operation types - Set and Remove
                if (batchAction.SetKeys != null)
                {
                    for (int i = 0; i < batchAction.SetKeys.Length; i++)
                    {
                        var partitionName = batchAction.SetPartitionNames[i];
                        String key = batchAction.SetKeys[i];
                        Object value = batchAction.SetValues[i];
                        var isolatedState = GetIsolatedState(partitionName);
                        isolatedState.SetDurableValue(key, value, "secondary", scour: true);
                        //Console.WriteLine("ApplyActionOnSecondary: Adding key {0} for partitionKey {1}", key, partitionKey);
                    }
                }

                if (batchAction.RemoveKeys != null)
                {
                    for (int i = 0; i < batchAction.RemoveKeys.Length; i++)
                    {
                        var partitionName = batchAction.RemovePartitionNames[i];
                        String key = batchAction.RemoveKeys[i];
                        //this.Logger.Write(Guid.Empty, "Secondary servicing Remove({0}) in batch {1}", key, sequenceNumber);
                        var isolatedState = GetIsolatedState(partitionName);
                        isolatedState.RemoveDurableValue(key);
                        //Console.WriteLine("ApplyActionOnSecondary: Removing key {0} for partitionKey {1}", key, partitionKey);
                    }
                }

                this.sequenceNumber = sequenceNumber;
                return;
            }

            // FAIL if we come here
            Utility.CodingError("Unknown action asked to be applied");
            throw new InvalidOperationException();
        }

        protected override byte[] GetCurrentState(out long sequenceNumber)
        {
            // Serializing the entire state and returning the current sequence number for copy
            lock (this)
            {
                sequenceNumber = this.sequenceNumber;

                this.Logger.Write(Guid.Empty, "ActorState: Copy - serializing HashTable with {0} seq", sequenceNumber);
                using (var ms = new MemoryStream())
                {
                    using (var writer = new BinaryWriter(ms))
                    {

                        writer.Write(this.m_stateHelper.PermaStore.Count); // Number of dictionaries contained in this transmission

                        foreach (var outerKvp in this.m_stateHelper.PermaStore)
                        {
                            writer.Write(outerKvp.Key); // the partitionName
                            writer.Write(outerKvp.Value.Count); // The number of items in this dictionary

                            // Write the assemblies and assembly references first.  
                            // They need to be present before any delegates can be properly deserialized.
                            foreach (var kvp in outerKvp.Value)
                            {

                                bool isAssembly = kvp.Key.StartsWith("$ASSEMBLY$");
                                bool isAssemblyReference = kvp.Key.StartsWith("$REF$ASSEMBLY$");

                                if (!(isAssembly || isAssemblyReference)) continue;
                                string key;
                                object value = null;
                                if (isAssembly)
                                {
                                    key = kvp.Key;
                                    value = kvp.Value;
                                    Contract.Assume(value != null, "We will never add an assembly with a null value.");
                                }
                                else
                                {
                                    // This is an assembly reference.  Convert it into an assembly.
                                    // (There is no guarantee that the assembly being referenced will be present
                                    // when this state is deserialized, so we need to serialize the assembly itself.)
                                    key = kvp.Key.Substring(4); // Lop off the "$REF" prefix;
                                    var hash = (string)kvp.Value;
                                    var cacheKey = key + hash;
                                    byte[] cachedAssemblyBytes = null;
                                    lock (AssemblyHelper.s_assemblyCache)
                                    {
                                        Contract.Assert(AssemblyHelper.s_assemblyCache.ContainsKey(cacheKey), "Trying to replicate an assembly reference that doesn't exist in our actor state.");
                                        cachedAssemblyBytes = AssemblyHelper.s_assemblyCache[cacheKey];
                                    }
                                    value = Tuple.Create(hash, cachedAssemblyBytes); // Tuple<string,byte[]>
                                }
                                Contract.Assert(value != null, "GetCurrentState found a null assembly value.  This makes no sense.");
                                writer.Write(key);
                                MemoryStream objStream = new MemoryStream();
                                binaryFormatter.Serialize(objStream, value);
                                writer.Write((int)objStream.Length);
                                writer.Flush();
                                objStream.WriteTo(ms);
                                //Console.WriteLine("GetCurrentState: Wrote key {0} ({2}) for partitionId {1}", kvp.Key, outerKvp.Key, kvp.Value);
                            }

                            // Now write the rest of the entries.
                            foreach (var kvp in outerKvp.Value)
                            {
                                bool isAssembly = kvp.Key.StartsWith("$ASSEMBLY$");
                                bool isAssemblyReference = kvp.Key.StartsWith("$REF$ASSEMBLY$");

                                if (isAssembly || isAssemblyReference) continue;
                                writer.Write(kvp.Key);
                                MemoryStream objStream = new MemoryStream();
								if (kvp.Value != null)
                                    binaryFormatter.Serialize(objStream, kvp.Value);
                                writer.Write((int)objStream.Length);
                                writer.Flush();
                                objStream.WriteTo(ms);
                                //Console.WriteLine("GetCurrentState: Wrote key {0} ({2}) for partitionId {1}", kvp.Key, outerKvp.Key, kvp.Value);
                            }

                        }
                        return ms.ToArray();
                    }
                }
            }
        }
                

        protected override void SetCurrentState(byte[] serializedState, long sequenceNumber)
        {
            // No need to lock - it is guaranteed that this is called serially
            this.sequenceNumber = sequenceNumber;
            //this.m_stateHelper.PermaStore.Clear();
            using (var ms = new MemoryStream(serializedState))
            {
                using (var reader = new BinaryReader(ms))
                {
                    this.Logger.Write(Guid.Empty, "ActorState: Copy - deserializing HashTable with {0} seq", sequenceNumber);

                    int outerCount = reader.ReadInt32(); // read number-of-dictionaries
                    for (int outer = 0; outer < outerCount; outer++)
                    {
                        string partitionName = reader.ReadString(); // partition key for this dictionary
                        int count = reader.ReadInt32(); // number of items for this dictionary
                        //m_stateHelper.PermaStore[partitionName] = new ConcurrentDictionary<string, object>();
                        IsolatedActorState isolatedState = GetIsolatedState(partitionName);
                        for (int i = 0; i < count; i++)
                        {
                            string key = reader.ReadString();
                            int valueLength = reader.ReadInt32();
                            object value = null;
                            if (valueLength > 0)
                            {
                                byte[] valueBytes = reader.ReadBytes(valueLength);
                                var valueStream = new MemoryStream(valueBytes, false);
                                value = binaryFormatter.Deserialize(valueStream);
                            }
                            isolatedState.SetDurableValue(key, value, "deserializer", scour: false);
                            //Console.WriteLine("SetCurrentState: Read key {0} ({2}) for partitionName {1}", key, partitionName, value);

                        }
                    }
                }
            }
        }

        #endregion StateProviderBase methods

        // Action taken when a "batch" is ready to be persisted/replicated
        async Task IPersistenceHandler.OnPersistAsync(IList<ChangeOperation> toPersist)
        {
            var action = new BatchAction();
            action.StoreActions(toPersist);

            Task<Object> lastReplicateTask = null;
            do
            {
                lastReplicateTask = this.ReplicateAsync(action);
                await lastReplicateTask.ConfigureAwait(continueOnCapturedContext: false);
                Contract.Assert(!lastReplicateTask.IsFaulted, "lastReplicateTask faulted and we didn't throw an exception.");
                action = action.NextAction;
            } while (action != null);
        }
    }

    [Serializable]
    public class BatchAction : ActionBase
    {
        public BatchAction()
        {
        }

        public string[] SetPartitionNames { get; set; }
        public string[] SetKeys { get; set; }
        public Object[] SetValues { get; set; }

        public string[] RemovePartitionNames { get; set; }
        public string[] RemoveKeys { get; set; }

        // Windows Fabric has a limit of 50 MB of data for a single replication request.  Let's chain BatchActions if we hit that limit.
        // However, we can't easily predict the serialized size of data.  For an actor list of Int32's, the binary serialized blob size
        // for 1 million Int32's is about 64 times bigger (for the name of the collection, the key name, the tuple of value and two indices, etc).
        // Let's round that number up a bit to account for Strings and other interesting types, perhaps to 200.  This is a guess at the
        // the maximum number of elements we should send in one batch.
        private const int MaxReplicationSizeInBytes = 50 << 20;
        private const int ApproximateMaxBatchActionItems = MaxReplicationSizeInBytes / 200;  // 200 bytes per serialized item?

        // For building a list of batch actions for replication.  The entire point though is to not serialize these together.
        // But we want a chain of them to keep our replication logic simpler.
        [NonSerialized]
        private BatchAction _nextAction;

        // If we use the JSON.Net serializer for replication, ensure we do not try to serialize this property.
        internal BatchAction NextAction {
            get { return _nextAction; }
        }

        // To minimize serialization costs, don't serialize ActionBase instances.
        // Instead, serialize just the values.  Saves about 30% on a microbenchmark.
        internal void StoreActions(IList<ChangeOperation> pending)
        {
            Contract.Requires(pending != null);

            StoreActionsHelper(pending, 0);
        }

        internal void StoreActionsHelper(IList<ChangeOperation> pending, int startIndex)
        {
            Contract.Requires(pending != null);
            Contract.Requires(startIndex >= 0);

            int numInThisChunk = Math.Min(pending.Count - startIndex, ApproximateMaxBatchActionItems);

            var newPartitionNames = new List<string>(numInThisChunk);
            var newKeys = new List<string>(numInThisChunk);
            var newValues = new List<Object>(numInThisChunk);
            var removePartitionNames = new List<string>();
            var removeKeys = new List<string>();

            for (int i = startIndex; i < numInThisChunk + startIndex; i++)
            {
                var op = pending[i];
                if (op.OperationType == ChangeOperationType.Set)
                {
                    newPartitionNames.Add(op.PartitionName);
                    newKeys.Add(op.OperationKey);
                    newValues.Add(op.OperationValue);
                }
                else if (op.OperationType == ChangeOperationType.Remove)
                {
                    removePartitionNames.Add(op.PartitionName);
                    removeKeys.Add(op.OperationKey);
                }
                else
                {
                    // If adding a new Action type, add it here as well as in ApplyActionOnPrimary & ApplyActionOnSecondary
                    Contract.Assume(false, "Unsupported ChangeOperation type in BatchAction.Store.");
                    throw new NotImplementedException("Unsupported ChangeOperation type in BatchAction.Store.");
                }
            }

            if (newKeys.Count > 0)
            {
                SetKeys = newKeys.ToArray();
                SetValues = newValues.ToArray();
                SetPartitionNames = newPartitionNames.ToArray();
            }
            if (removeKeys.Count > 0)
            {
                RemoveKeys = removeKeys.ToArray();
                RemovePartitionNames = removePartitionNames.ToArray();
            }

            if (pending.Count > startIndex + numInThisChunk)
            {
                _nextAction = new BatchAction();
                _nextAction.StoreActionsHelper(pending, startIndex + numInThisChunk);
            }
        }
    }


}
